package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

public class ExercisePv1 {
    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        //输入数据
        SingleOutputStreamOperator<UserBehavior> stream = env.readTextFile("data/file/UserBehavior.csv").map(ads -> {
            String[] line = ads.split(",");
            return new UserBehavior(Long.valueOf(line[0]), Long.valueOf(line[1]), Integer.valueOf(line[2]), line[3], Long.valueOf(line[4]));
        }).filter(x -> x.getBehavior().equals("pv"));
        WatermarkStrategy<UserBehavior> strategy = WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((user, ts) -> user.getTimestamp());
        //8.1.1	指定时间范围内网站总浏览量（PV）的统计
        mc2(strategy,stream);

        //mc1(strategy,stream);

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void mc2(WatermarkStrategy<UserBehavior> strategy, SingleOutputStreamOperator<UserBehavior> stream) {
        stream.assignTimestampsAndWatermarks(strategy).map(x-> Tuple2.of("pv",1L))
                .returns(Types.TUPLE(Types.STRING,Types.LONG)).keyBy(t->t.f0)
                .window(TumblingEventTimeWindows.of(Time.hours(1))).sum(1)
                .print();
    }

    private static void mc1(WatermarkStrategy<UserBehavior> strategy,SingleOutputStreamOperator<UserBehavior> stream) {
        stream
                .assignTimestampsAndWatermarks(strategy)
                .keyBy(UserBehavior::getBehavior)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0l;
                    }

                    @Override
                    public Long add(UserBehavior value, Long acc) {
                        return acc + 1;
                    }

                    @Override
                    public Long getResult(Long acc) {
                        return acc;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        System.out.println("-----");
                        return a + b;
                    }
                }, new ProcessWindowFunction<Long, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context ctx, Iterable<Long> elements, Collector<String> out) throws Exception {

                        for (Long ele : elements) {
                            System.out.println(ele);
                        }
                        Long result = elements.iterator().next();
                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm");
                        String start = format.format(new Date(ctx.window().getStart()));
                        String end = format.format(new Date(ctx.window().getEnd()));
                        String msg="key: "+key+" start: "+start+" end: "+end+ " result: "+result;
                        out.collect(msg);

                    }
                })
                .print();
    }

}
